package state;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import scala.Int;

import java.io.IOException;

public class state {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.fromElements(
                Tuple2.of(1,1),
                Tuple2.of(1,2),
                Tuple2.of(1,3),
                Tuple2.of(1,4),
                Tuple2.of(1,5),
                Tuple2.of(1,6),
                Tuple2.of(1,7)
        ).keyBy(0)
                .flatMap(new richmap())
                .print()
                .setParallelism(1);

        env.execute("countstate");

    }

}

// 创建一个类
class richmap extends RichFlatMapFunction<Tuple2<Integer,Integer>,Tuple2<Integer,Integer>>{


    //定义一个状态值tuple2 ， 第一个值存位数，计算到哪一位了，第一个值存值，计算的和是多少了
    private ValueState<Tuple2<Integer,Integer>> state;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化状态
        ValueStateDescriptor<Tuple2<Integer, Integer>> descriptor = new ValueStateDescriptor<Tuple2<Integer, Integer>>(
                "flamapstate", // 状态名字
                TypeInformation.of(new TypeHint<Tuple2<Integer, Integer>>() {}),//状态的类型
                Tuple2.of(0, 0) // 状态的初始值
        );
        //super.open(parameters);
        state = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void flatMap(Tuple2<Integer, Integer> integerIntegerTuple2, Collector<Tuple2<Integer, Integer>> collector) throws IOException {
    // 获取当前状态
        Tuple2<Integer, Integer> currentstate = state.value();
    //    执行累加并计算
        currentstate.f0 ++; // 位数加1
        currentstate.f1 = currentstate.f1 + integerIntegerTuple2.f1; // 值求和

    //    更新状态
        state.update(currentstate);

    //    如果达到第三个数字就重新数位数
        if(currentstate.f0>=3){
            collector.collect(new Tuple2<Integer, Integer>(integerIntegerTuple2.f0,currentstate.f1));
        //    清空状态重新计算 todo 这里感觉不应该情况状态吧
            state.clear();
        }


    }
}
